package org.raftkv.raft.pb;

import org.raftkv.commons.Tuple;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.enums.ConfChangeType;
import org.raftkv.raft.pb.enums.EntryType;
import org.raftkv.raft.pb.enums.MessageType;
import org.raftkv.raft.pb.impl.ConfChange;
import org.raftkv.raft.pb.impl.ConfChangeV2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 该接口对 ConfChangeV2 和 ConfChange 进行了抽象，允许以统一的方式处理它们。
 *
 * @author wzy
 */
public interface IConfChange {

    Logger LOGGER = LoggerFactory.getLogger(IConfChange.class);

    ConfChangeV2 asV2();

    Tuple<ConfChange, Boolean> asV1();

    static List<ConfChangeSingle> confChangesFromString(String s) {
        List<ConfChangeSingle> ccs = new ArrayList<>();
        String[] toks = s.trim().split(" ");
        if ("".equals(toks[0])) {
            return null;
        }

        for (String tok : toks) {
            if (tok.length() < 2) {
                String errMsg = String.format("unknown token %s", tok);
                LOGGER.error(errMsg);
                throw new RaftException(errMsg);
            }

            ConfChangeSingle cc = new ConfChangeSingle();
            switch (tok.substring(0, 1)) {
                case "v":
                    cc.setType(ConfChangeType.ConfChangeAddNode);
                    break;
                case "l":
                    cc.setType(ConfChangeType.ConfChangeAddLearnerNode);
                    break;
                case "r":
                    cc.setType(ConfChangeType.ConfChangeRemoveNode);
                    break;
                case "u":
                    cc.setType(ConfChangeType.ConfChangeUpdateNode);
                    break;
                default:
                    String errMsg = String.format("unknown input: %s", tok);
                    LOGGER.error(errMsg);
                    throw new RaftException(errMsg);
            }
            long id = Long.parseLong(tok.substring(1));
            cc.setNodeId(id);
            ccs.add(cc);
        }

        return ccs;
    }

    static String confChangesToString(List<ConfChangeSingle> ccs) {
        StringBuilder buf = new StringBuilder();
        for (int i = 0; i < ccs.size(); i++) {
            ConfChangeSingle cc = ccs.get(i);
            if (i > 0) {
                buf.append(" ");
            }
            switch (cc.getType()) {
                case ConfChangeAddNode:
                    buf.append('v');
                    break;
                case ConfChangeAddLearnerNode:
                    buf.append('l');
                    break;
                case ConfChangeRemoveNode:
                    buf.append('r');
                    break;
                case ConfChangeUpdateNode:
                    buf.append('u');
                    break;
                default:
                    buf.append("unknown");
            }
            buf.append(cc.getNodeId());
        }
        return buf.toString();
    }

    static Message confChangeToMsg(IConfChange c) {
        Tuple<EntryType, byte[]> t = marshalConfChange(c);
        EntryType typ = t.getFirst();
        byte[] data = t.getSecond();
        List<Entry> es = new ArrayList<>();
        Entry e = new Entry();
        e.setData(data);
        e.setType(typ);
        es.add(e);
        return new Message.MessageBuilder()
                .type(MessageType.MsgProp)
                .entries(es)
                .build();
    }

    private static Tuple<EntryType, byte[]> marshalConfChange(IConfChange c) {
        EntryType typ;
        byte[] ccdata;

        Tuple<ConfChange, Boolean> temp = c.asV1();
        ConfChange ccv1 = temp.getFirst();
        Boolean ok = temp.getSecond();
        if (ok) {
            typ = EntryType.EntryConfChange;
            ccdata = ccv1.marshal();
        } else {
            ConfChangeV2 ccv2 = c.asV2();
            typ = EntryType.EntryConfChangeV2;
            ccdata = ccv2.marshal();
        }
        return new Tuple<>(typ, ccdata);
    }

}
